package com.service;

import com.config.CommandParse;
import com.databus.DatabusConnector;
import com.databus.Log;
import com.databus.PublishMapping;
import com.databus.RequestMapping;
import com.google.protobuf.GeneratedMessage;
import com.packet.PackageData;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

public  class Consumer extends DatabusConnector implements DatabusConnector.OnPublishListener,DatabusConnector.OnConnectChangedListener {

    private ConcurrentHashMap<Class,Method> publishMap=new ConcurrentHashMap<>();

    public Consumer()
    {
        this.addConnectChangedListener(this);
        this.addPublishListener(this);

        Class clazz=this.getClass();
        Method[] methods=clazz.getMethods();
        for(Method method:methods){
            PublishMapping pubMapping=method.getAnnotation(PublishMapping.class);
            if(pubMapping!=null && method.getParameterCount()==1)
            {
                Class[] types=method.getParameterTypes();
                publishMap.put(types[0],method);
            }
        }
    }

    private void AddFunction(List<String> functionlist,Class clazz)
    {
        String typename=clazz.getCanonicalName();
        String packname=clazz.getPackage().getName();
        if(packname.length()>0)
            typename=typename.replace(packname+".","");
        functionlist.add(typename);
    }

    private String getSimpleCanonicalName(Class clazz)
    {
        String simplename=clazz.getSimpleName();
        String typename=clazz.getCanonicalName();
        String packname=clazz.getPackage().getName();
        if(packname.length()>0)
            typename=typename.replace(packname+".","");
        return typename;
    }

    @Override
    public List<Integer> getTopics()
    {
        List<Integer> topics = new ArrayList<>();
        for(Class clazz:publishMap.keySet()){
            if(GeneratedMessage.class.isAssignableFrom(clazz))
            {
                String typename=getSimpleCanonicalName(clazz);
                topics.add(CommandParse.Hashcode(typename));
            }
        }
        return topics;
    }

    @Override
    public void onPublish(PackageData msg) {
        //System.out.println(msg.toString());
        Class clazz=msg.mMsg.getClass();
        Object resp=null;
        if(publishMap.containsKey(clazz))
        {
            Method method=publishMap.get(clazz);
            try {
                resp=method.invoke(this,msg.mMsg);
            } catch (Exception e) {
                Log.error(e);
            }
        }
    }

    @Override
    public void onConnectChanged(EConnectStatus status) {

    }
}